package com.dctp.common.disruptor;

import com.dctp.common.core.RedisManager;
import com.dctp.common.service.TradeService;
import com.dctp.common.utils.DateUtil;
import com.dctp.common.vo.FundChangeType;
import com.dctp.common.vo.FundFlowVo;
import com.jfinal.plugin.activerecord.DbKit;
import com.jfinal.plugin.activerecord.Record;
import com.lmax.disruptor.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

public class FundFlowEventHandler implements EventHandler<FundFlowEvent> {

    private static final Logger logger = LoggerFactory.getLogger(FundFlowEventHandler.class);
    private final List<FundFlowVo> batchBuffer = new ArrayList<>(200);
    private final int batchSize = 50;  // 每批写入条数

    @Override
    public void onEvent(FundFlowEvent event, long sequence, boolean endOfBatch) throws Exception {
        if (event.getOrder() != null) {
            batchBuffer.add(event.getOrder());
        }

        // ✅ 满 50 条 或 Disruptor 告诉我「一批结束」就写库
        if (batchBuffer.size() >= batchSize || endOfBatch) {
            try {
                saveBatchFundFlow(batchBuffer);
                logger.info("✅ 批量写入订单 {} 条", batchBuffer.size());
            } catch (Exception e) {
                logger.error("❌ 批量写订单失败", e);
                // TODO: 失败重试 or 落地文件
            } finally {
                batchBuffer.clear();
            }
        }
    }

    public void saveBatchFundFlow(List<FundFlowVo> fundFlowList) throws SQLException {
        String sql = "INSERT INTO dctp_fund_flow (" +
                "user_id, balance_before, change_amount, balance_after, change_type, " +
                "order_id, remark, parent_id, bottom_role_user_id, middle_role_user_id, top_role_user_id, " +
                "create_time, update_time" +
                ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)";

        // ✅ 获取连接（可与 saveBatchOrders 共享连接，性能更好）
        try (Connection conn = DbKit.getConfig().getConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {

            conn.setAutoCommit(false);  // 🚀 一次事务提交

            LocalDateTime now = DateUtil.now();
            for (FundFlowVo flow : fundFlowList) {
                int i = 1;
                ps.setLong(i++, flow.getUserId());
                ps.setBigDecimal(i++, flow.getBalance());
                ps.setBigDecimal(i++, flow.getChangeAmount());
                ps.setBigDecimal(i++, flow.getBalanceAfter());
                ps.setString(i++, flow.getCode().toString());  // FundChangeType 转 string

                ps.setLong(i++, flow.getOrderId());
                ps.setString(i++, flow.getRemark());
                ps.setObject(i++, flow.getParentId());
                ps.setObject(i++, flow.getBottomRoleUserId());
                ps.setObject(i++, flow.getMiddleRoleUserId());
                ps.setObject(i++, flow.getTopRoleUserId());

                ps.setObject(i++, now);
                ps.setObject(i++, now);

                ps.addBatch();
            }

            ps.executeBatch();
            conn.commit();  // 🚀 提交事务
        }
    }
}
